#include "../util/h/iUtil.h"
#include "../util/h/iLogger.h"
#include "iListenConnection.h"
#include "iSocket.h"
#include "iOTCPConnection.h"
#include "iPBNConnection.h"
#include "../util/src/iIOThread.h"

iTPListener* iTPListener::new_instance(const char* address, iAllocator* alloc)
{
    FUNC("iTPListener::new_instance");

    char proto[32],ip[64],port[8];
    if(!parse_address(address, proto, ip, port))
        return NULL;

    INFO("create listener on address="<<address);
	if(strstr(proto, "PKG") != NULL)
		return new(alloc->allocate(sizeof(iPBNListenConnection))) iPBNListenConnection(address, alloc);
	else
		return new(alloc->allocate(sizeof(iListenConnection))) iListenConnection(address, alloc);
}

iListenConnection::iListenConnection(const char* address, iAllocator* alloc)
:iConnection(NULL, alloc)
,m_sink(NULL)
,m_soctx(NULL)
{
	m_soctx = soctx::new_instance(address);
	m_jobber = (iMsgThread*)iThreadMgr::spawn_this();
	m_worker = (iMsgThread*)iThreadMgr::spawn_io();
}

bool iListenConnection::listen(iTPListener::iSink* sink)
{
	FUNC("iListenConnection::listen");

	if(m_sink)
		return false;

	m_sink = sink;

	if(NULL == m_sink)
		return false;

	m_socket = new iListenSocket(m_soctx);
	
	if(!m_socket->bind(*m_soctx) || ((iListenSocket*)m_socket)->listen())
		return false;

	m_socket->set_sink(this);

	((iIOThread*)m_worker)->post_msg(iMsg::new_instance(MSG_TP_LISTEN, NULL ,this), m_jobber);
	return true;
}

void iListenConnection::close()
{
	FUNC("iListenConnection::close");
}

void iListenConnection::on_connect_indication(iClientSocket* cs)
{   
	FUNC("iListenConnection::on_connect_indication");

	if(NULL == cs)
		return;

    iOTCPConnection* conn = new iOTCPConnection(cs,this, iSingleton<iAllocator>::instance());
	conn->m_conn_status = iConnection::CS_ESTABLISHED;
    conn->welding(m_jobber, m_worker);
    cs->set_sink(conn);
    if(m_soctx->get_protocol() == soctx::TCP)
        ((iIOThread*)m_worker)->add_io(cs);
    struct payload {iTPConnection* conn; iMsgDB* request;} *pl = new payload;
    pl->conn = conn;
    pl->request = NULL;
    ((iMsgThread*)m_jobber)->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_IN, pl, this), m_worker);
}
int iListenConnection::on_message(iMsgID msg, void* param)
{
	switch(msg)
	{
    case MSG_TP_CONNECTION_IN:
        {
            struct payload {iTPConnection* conn; iMsgDB* request;} *pl = (payload*)param;
            m_sink->on_connection_indication(pl->conn, pl->request);
        }
        break;
    case MSG_TP_LISTEN:
        {
            DEBUG("join IO processor");
            ((iIOThread*)m_worker)->add_io(m_socket);
        }
        break;
	}

	return 0;
}
/************************************************************************/
/*    iPBNListenConnection                                              */
/************************************************************************/
iPBNListenConnection::iPBNListenConnection(const char* address, iAllocator* alloc)
:iListenConnection(address, alloc)
{

}
void iPBNListenConnection::on_connect_indication(iClientSocket* cs)
{
	FUNC("iPBNListenConnection::on_connect_indication");

	iPBNConnection* conn = new iPBNConnection(cs, this, iSingleton<iAllocator>::instance());
	cs->set_sink(conn);
	conn->welding(m_jobber, m_worker);
    if(m_soctx->get_protocol() == soctx::TCP)
        ((iIOThread*)m_worker)->add_io(cs);
	conn->on_connect_confirm();
}
int iPBNListenConnection::on_message(iMsgID msg, void* param)
{
    return iListenConnection::on_message(msg, param);
}	
